#include "app_pool.h"
#include <pthread.h>
#include <mqueue.h>
#include "log/log.h"
#include <stdlib.h>

static char *mq_name = "/app_pool_queue";
static mqd_t mq_desc;
static int threads_count;
static pthread_t *threads;

static void *thread_fun(void *arg)
{
    Task task;
    while (1)
    {
        ssize_t len = mq_receive(mq_desc, (char *)&task, sizeof(Task), NULL);
        if (len == sizeof(Task))
        {
            task.task_fun(task.arg);
        }
    }
}

/**
 * @brief 初始化
 *
 * @param num 线程池线程数量
 * @return int 判断是否初始化成功
 */
int app_pool_init(int num)
{
    // 创建任务队列
    struct mq_attr attr;
    attr.mq_msgsize = sizeof(Task);
    attr.mq_maxmsg = 5;
    mq_desc = mq_open(mq_name, O_CREAT | O_RDWR, 0644, &attr);
    if (mq_desc == -1)
    {
        log_error("创建任务队列失败!");
        return -1;
    }

    // 创建线程池
    threads_count = num;
    threads = (pthread_t *)malloc(num * sizeof(pthread_t));
    for (int i = 0; i < num; i++)
    {
        pthread_create(threads + i, NULL, thread_fun, NULL);
    }
    return 0;
}

/**
 * @brief 关闭所有
 *
 */
void app_pool_close(void)
{
    mq_close(mq_desc);
    mq_unlink(mq_name);

    for (int i = 0; i < threads_count; i++)
    {
        pthread_cancel(threads[i]);
        pthread_join(threads[i], NULL);
    }
    free(threads);
}

/**
 * @brief 注册回调函数
 *
 * @param task_fun 回调函数
 * @param arg 回调函数参数
 */
void app_pool_registerCallback(int (*task_fun)(void *), void *arg)
{
    Task task = {
        .task_fun = task_fun,
        .arg = arg,
    };

    mq_send(mq_desc, (char *)&task, sizeof(Task), 0);
}
